摘要
這篇文章探討了 多代理系統 在人工智能領域的應用，特別是 如何利用 LangGraph 框架來構建和管理多代理系統。文章首先介紹了多代理系統的基本概念，包括 由多個獨立的 AI 行為者（Agents）組成，每個行為者擁有自己的「大腦」和「工具箱」。接著，文章重點闡述了兩種常見的多代理設計模式：協作模式 (Collaboration) 和監督者模式 (Supervisor)。協作模式強調 將複雜任務分解成可處理的工作單元，由不同的專門代理負責處理，而監督者模式則 引入一個中央控制單元來協調和指揮其他專業代理的工作。
文章以 多代理協作翻譯系統 為例，展示了如何利用 LangGraph 框架來實作協作模式。整個翻譯系統包含了 多個 AI 代理，分別負責生成翻譯、評估翻譯、深入分析和最終排名。文章詳細說明了每個代理的功能和流程，並展示了如何使用 LangGraph 的 Send() 函數來實現 Map-Reduce 模式，讓系統能高效地並行處理多個翻譯版本。最後，文章總結了多代理系統的優勢和潛在應用，鼓勵讀者利用 LangGraph 框架來構建和管理更複雜的 AI 系統。
在人工智能快速發展的今天，單一代理（Agent）系統已經無法滿足日益複雜的任務需求。多代理系統應運而生，為我們提供了更強大、更靈活的問題解決方案。本文將深入探討多代理系統的核心概念、優勢，以及兩種常見的設計模式：協作模式和監督者模式。我們還將通過實際案例，展示如何運用 LangGraph 框架來實現這些模式。
多代理系統是由多個獨立的 AI 行為者（Agents）組成的網絡，這些行為者由語言模型驅動，並以特定方式相互連接。
💡 核心概念：每個代理都是一個獨立的 AI 實體，擁有自己的「大腦」（語言模型）和「工具箱」（自定義程式碼）。
關鍵組成要素
獨立代理：每個 Agent 可以擁有：
連接機制：定義代理間如何協作和通訊。
面對複雜任務時，多代理系統相較於單一代理有顯著優勢：
多代理設計允許你將複雜的問題分解成可處理的工作單元，這些單元可以由專門的代理和語言模型程序來處理。
單一代理通常可以在單一領域內有效地使用少數工具，儘管如 GPT-4 這樣的尖端大型語言模型（LLM）在單一領域內表現出色，但面對跨領域的複雜任務時，其效能往往難以充分發揮。
處理複雜任務的一種方法是採用「分而治之」的方法：為每個任務或領域創建一個專門的代理，並將任務路由到正確的「專家」。具體而言，它為每個子任務或專業領域配置一個專門的 AI 代理，形成一個協同運作的智能網絡。
最近，Wu 等學者在其開創性論文《AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation》中，詳細闡述了這一概念。而 LangGraph 框架則提供了將此理論付諸實踐的強大工具。通過 LangGraph，開發者可以輕鬆構建和管理複雜的多代理系統，實現任務的智能分發和協作處理。
監督者模式的核心思想是引入一個中央控制單元——監督者代理（Supervisor Agent）。這個監督者就像是一個經驗豐富的項目經理，負責協調和指揮其他專業代理的工作。
💡 核心概念：監督者代理利用語言模型的強大能力，動態分配任務並管理整個代理團隊的工作流程。
監督者模式代表了一種革命性的任務處理方法。核心是一個中央AI監督代理，它負責接收並分析複雜任務，制定執行策略，動態分配子任務給專業代理，實時監控進度，並最終整合結果。這種模式模仿了高效團隊的工作方式，能夠處理更複雜的問題，提高系統的靈活性和智能程度。
在這個專案中，我們利用 LangGraph 框架構建了一個高效的多代理協作翻譯系統。該系統的核心理念是通過多個專業 AI 代理的協作，實現從初步翻譯到最終優化的完整流程。這種方法不僅提高了翻譯的質量，還為翻譯決策提供了深入的見解和理由。
系統的工作流程包括以下幾個關鍵步驟：
這種多步驟的方法允許我們充分利用 AI 模型的能力，同時通過多層次的評估和分析來確保翻譯的高質量。
在開始實際的程式碼實現之前，讓我們說明這個多代理協作翻譯系統的設計流程。這個系統模擬了一個高效的團隊合作過程，每位成員都有其專門的職責。
以下是系統的詳細設計流程,每個組件都扮演著關鍵角色:
發派員（Brainstorm Translation）：
Send() 函數將任務分發給多個翻譯方向翻譯方向（Generation Translation）：
評估人員（Evaluate Translation）：
深思人員（Deep Thought Translation）：
裁判（Ranked Translation）：
想像我們要翻譯的句子是 "台灣是個美好的地方"。
首先，我們需要導入必要的庫和模組：
import operator
from typing import Annotated, TypedDict
from langchain_core.pydantic_v1 import BaseModel
from langgraph.constants import Send
from langgraph.graph import END, StateGraph, START
from langchain_openai import ChatOpenAI
這些導入為我們提供了構建複雜工作流所需的工具和類型。
我們首先定義了系統中的關鍵節點，每個節點代表翻譯流程中的一個重要步驟：
GENERATE_TRANSLATIONS = "GENERATE_TRANSLATIONS"
EVALUATE_TRANSLATION = "EVALUATE_TRANSLATION"
DEEP_ANALYZE = "DEEP_ANALYZE"
RANK_TRANSLATIONS = "RANK_TRANSLATIONS"
為了有效管理整個翻譯過程中的數據流，我們定義了兩個關鍵的狀態結構：
OverallState -
"原文":
"譯文": ["版本", "版本", "版本"],
"評估": ["結果“, "結果“, "結果“],
"深思": ["結果“, "結果“, "結果“],
"翻譯排名": [""]
class OverallState(TypedDict):
input: str
translations: Annotated[list[str], operator.add]
reviews: Annotated[list[str], operator.add]
deep_analyses: Annotated[list[str], operator.add]
ranked_translations: str
class TranslationState(TypedDict):
translation: str
OverallState 用於跟踪整個過程的狀態，而 TranslationState 則用於單個翻譯的處理。
這個組件負責生成多個不同的翻譯版本：
class Translations(BaseModel):
translations: list[str]
model = ChatOpenAI(model="gpt-3.5-turbo-0125")
step1_prompt = """Step 1: Given the original English sentence: "{input}", please brainstorm three distinct translation options in Traditional Chinese. Consider factors such as accuracy, fluency, cultural appropriateness, and idiomatic expression."""
def generate_translations(state: OverallState):
prompt = step1_prompt.format(input=state["input"])
response = model.with_structured_output(Translations).invoke(prompt)
return {"translations": response.translations}
評估器對每個翻譯版本進行品質評估：
class Review(BaseModel):
review: str
step2_prompt = """Step 2: For each of the three proposed translations, evaluate their quality. Consider their accuracy, fluency, cultural appropriateness, and overall effectiveness. Assign a score out of 10 for each aspect.
Translations:
{translations}"""
def evaluate_translation(state: TranslationState):
prompt = step2_prompt.format(translations=state["translation"])
response = model.with_structured_output(Review).invoke(prompt)
return {"reviews": [response.review]}
深度分析器提供更細緻的翻譯分析：
class DeepAnalysis(BaseModel):
deep_analysis: str
step3_prompt = """Step 3: For each translation, provide a deeper analysis. Consider how well it captures the original meaning, its suitability for the target audience, and any cultural nuances. Suggest potential improvements or alternative phrasings.
Evaluation:
{review}"""
def deep_analyze(state: TranslationState):
prompt = step3_prompt.format(review=state["translation"])
response = model.with_structured_output(DeepAnalysis).invoke(prompt)
return {"deep_analyses": [response.deep_analysis]}
最後，排名器綜合所有信息，對翻譯進行最終排序：
class RankedTranslations(BaseModel):
ranked_translations: str
step4_prompt = """Step 4: Based on the evaluations and analyses, rank the translations in order of overall quality. Provide a justification for each ranking and offer any final thoughts or considerations for improving the best translation. Please Display Result in Traditional Chinese
Detailed analysis:
{deep_analysis}"""
def rank_translations(state: OverallState):
deep_analyses = "\n\n".join(state["deep_analyses"])
prompt = step4_prompt.format(deep_analysis=deep_analyses)
response = model.with_structured_output(RankedTranslations).invoke(prompt)
return {"ranked_translations": response.ranked_translations}
為了將這些組件整合成一個連貫的工作流程，我們使用 LangGraph 的 StateGraph 來定義節點之間的連接和數據流：
def continue_to_evaluation(state: OverallState):
return [Send(EVALUATE_TRANSLATION, {"translation": t}) for t in state["translations"]]
def continue_to_deep_analysis(state: OverallState):
return [Send(DEEP_ANALYZE, {"translation": r}) for r in state["reviews"]]
graph = StateGraph(OverallState)
graph.add_node(GENERATE_TRANSLATIONS, generate_translations)
graph.add_node(EVALUATE_TRANSLATION, evaluate_translation)
graph.add_node(DEEP_ANALYZE, deep_analyze)
graph.add_node(RANK_TRANSLATIONS, rank_translations)
graph.add_edge(START, GENERATE_TRANSLATIONS)
graph.add_conditional_edges(GENERATE_TRANSLATIONS, continue_to_evaluation, [EVALUATE_TRANSLATION])
graph.add_conditional_edges(EVALUATE_TRANSLATION, continue_to_deep_analysis, [DEEP_ANALYZE])
graph.add_edge(DEEP_ANALYZE, RANK_TRANSLATIONS)
graph.add_edge(RANK_TRANSLATIONS, END)
app = graph.compile()
最後，我們可以通過以下方式執行整個翻譯流程：
import rich
for s in app.stream({
"input": "The early bird catches the worm.",
}):
rich.print(s)
我們已經成功實現了多代理協作翻譯系統，但你可能好奇：這個系統如何實現高效的並行處理？答案藏在 LangGraph 框架的 Map-Reduce 模式實現中。
關鍵在於 Send() 函數，它在我們的程式碼中反覆出現。這個函數如何在 LangGraph 中實現 Map-Reduce 模式，從而支撐起整個系統的數據流動和並行處理？接下來，我們將深入探討 LangGraph 的 Map-Reduce 機制，揭示它如何使我們的協作翻譯系統高效運作。
Map-recude 是種常見的高效任務拆解操作。這種方法通常將任務拆解成較小的子任務，待子任務都完成後，整合所有子任務成果。主要分成兩步驟：Map 和 Reduce:
例如說在日常家庭大掃除，就可以用 map& reduce 來分析
在 LangGraph 的架構中，Send API 實現了 MapReduce 模式，透過條件邊緣（conditional edges）將不同的狀態分發到同一節點的多個實例上。
在 LangGraph 框架中，Map-Reduce 模式的實現主要依賴於 Send() 函數。這個函數是實現條件邊緣（conditional edges）的關鍵，允許我們將不同的狀態分發到同一節點的多個實例上。這種機制使得 Map-Reduce 模式在 LangGraph 中得以高效實現。
Send 的使用：Send 在實現 Map-Reduce 模式中扮演著重要角色。
a. continue_to_evaluation 函數：
- 這個函數將生成的多個翻譯版本分別發送到評估階段。
- 它為每個翻譯創建一個新的任務，確保每個翻譯都被獨立評估。
b. continue_to_deep_analysis 函數：
- 類似地，這個函數將每個評估結果發送到深入分析階段。
- 它確保每個翻譯的評估都能得到詳細的文化和語言分析。
節點函數：
generate_translations：生成多個可能的翻譯版本。evaluate_translation：評估單個翻譯的質量。deep_analyze：深入分析翻譯的文化適應性和語言細節。rank_translations：根據前面的分析對所有翻譯進行最終排名。這種結構允許我們同時處理多個翻譯版本，充分利用 LLM 的能力來生成多樣化的翻譯選項，並通過多步驟的評估和分析來選出最佳翻譯。這種方法不僅提高了翻譯的質量，還能為翻譯決策提供深入的見解和理由。
這種方法允許我們充分利用 AI 模型的能力，生成多樣化的翻譯選項，並通過多步驟的評估和分析來選出最佳翻譯。
通過這個項目，我們展示了如何利用 LangGraph 框架構建一個複雜的 AI 驅動翻譯系統。這種方法不僅能夠生成高質量的翻譯，還能提供深入的分析和持續改進的機會。對於需要處理大量文本翻譯任務的組織來說，這種系統可以大大提高效率和質量。
在實際應用中，您可以根據具體需求進一步優化每個步驟的提示詞，或者擴展系統以處理更多語言對。同時，考慮添加人工審核環節，可以在保證自動化效率的同時，確保翻譯的最終品質。
即刻前往教學程式碼 Repo，親自動手體驗多位 AI 代理協作的魅力吧！別忘了給專案按個星星並持續關注更新，讓我們一起探索AI代理的新境界。
這次準備概念卡片備妥在下方留言處，嘗試看看這種工作流
| 名詞 | 解釋 | 圖卡 |
|---|---|---|
| 多代理系統 | 由多個獨立 AI 代理組成的系統，適合處理複雜任務。 |  |
| 協作模式 | 多個代理分工合作，將大型任務分解並各自處理部分工作。 |  |
| 監督者模式 | 一個中央代理負責協調和指揮其他代理進行工作。 |  |
iThome鐵人賽
看影片追技術